package com.zzyy.study.mq;

import com.zzyy.study.lock.DistributeLock;
import com.zzyy.study.mq.consumer.retry.DefaultMsgConsumeRetry;
import com.zzyy.study.mq.consumer.retry.MsgConsumeRetry;
import com.zzyy.study.mq.mapper.MsgMapper;
import com.zzyy.study.mq.properties.MqProperties;
import com.zzyy.study.mq.sender.retry.DefaultMqSendRetry;
import com.zzyy.study.mq.sender.retry.MqSendRetry;
import com.zzyy.study.mq.sender.DefaultMsgSender;
import com.zzyy.study.mq.sender.IMsgSender;
import com.zzyy.study.mq.sender.MqSendRetryJob;
import com.zzyy.study.mq.service.IMsgService;
import com.zzyy.study.mq.service.impl.MsgServiceImpl;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import com.zzyy.study.mq.delay.DelayQueueTaskProcessor;

/**
 * @auther zzyy
 * @create 2024-07-21 12:10
 */
@Configuration
@ComponentScan
@MapperScan(basePackageClasses = MsgMapper.class)
@EnableScheduling
@EnableConfigurationProperties(MqProperties.class)
public class MqConfiguration {

    private MqProperties mqProperties;

    public MqConfiguration(MqProperties mqProperties) {
        this.mqProperties = mqProperties;
    }

    @Bean
    @ConditionalOnMissingBean
    public MqSendRetry mqSendRetry() {
        return new DefaultMqSendRetry();
    }

    @Bean
    public DelayQueueTaskProcessor delayMsgProcessor() {
        return new DelayQueueTaskProcessor("delayMsgProcessor", this.mqProperties.getDelayMsgDelayQueueCapacity(), Runtime.getRuntime().availableProcessors());
    }

    @Bean
    public DelayQueueTaskProcessor delaySendRetryProcessor() {
        return new DelayQueueTaskProcessor("delaySendRetryProcessor", this.mqProperties.getDelaySendRetryDelayQueueCapacity(), Runtime.getRuntime().availableProcessors());
    }

    @Bean
    @ConditionalOnMissingBean
    public IMsgSender msgSender(IMsgService msgService, MqSendRetry mqSendRetry, DistributeLock distributeLock, RabbitTemplate rabbitTemplate) {
        return new DefaultMsgSender(this.mqProperties,
                msgService,
                mqSendRetry,
                this.mqExecutor(),
                this.delayMsgProcessor(),
                this.delaySendRetryProcessor(),
                distributeLock,
                rabbitTemplate);
    }

    @Bean
    public IMsgService msgService() {
        return new MsgServiceImpl();
    }

    @Bean
    public MqSendRetryJob mqSendRetryJob(IMsgService msgService, IMsgSender msgSender) {
        return new MqSendRetryJob(msgService, msgSender);
    }

    @Bean
    public ThreadPoolTaskExecutor mqExecutor() {
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        threadPoolTaskExecutor.setThreadNamePrefix("ThreadPool-mqExecutor-");
        threadPoolTaskExecutor.setCorePoolSize(Runtime.getRuntime().availableProcessors() * 4);
        threadPoolTaskExecutor.setMaxPoolSize(Runtime.getRuntime().availableProcessors() * 8);
        threadPoolTaskExecutor.setQueueCapacity(10000);
        threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return threadPoolTaskExecutor;
    }

    @Bean
    @ConditionalOnMissingBean
    public MsgConsumeRetry msgConsumeRetry() {
        return new DefaultMsgConsumeRetry();
    }

}

